package com.hadoop.avg;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.lib.ChainMapper;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class AvgExe extends Configured implements Tool{

	
	
	//hadoop jar packagename.jar packagename.MainFunction <input_folder> <output_folder>
	//hadoop jar packagename.jar packagename.MainFunction -D workType=both <input_folder> <output_folder>
	/**
	 * @param args
	 * @throws IOException 
	 * @throws InterruptedException 
	 * @throws ClassNotFoundException 
	 */
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new AvgExe(), args);
		System.out.println("##Result : " + res);
	}

	@Override
	public int run(String[] arg0) throws Exception {
		Configuration conf = getConf();
		// 예를들어 -D myproperty=10 input output 이라고 입력하면 -D같은 옵션을 붙여서 설정한 파라미터는 모두 제외되고
		// input output만 return 됨.
		String[] otherArgs = new GenericOptionsParser(conf, arg0).getRemainingArgs();
		
		if(otherArgs.length != 2){
			System.err.println("Usage : Avg Data <Input> <Output>");
			System.exit(2);
		}
		
		if(conf.get("workType").equals("sum")){
			System.out.println("Wow~ To sum data, Wait kkk.. melong~");
			return 0;
		}
		
		//Job 이름 설정 
		Job job = new Job(conf, "Avg Price");
		
		//입출력 데이터 경로 설정
		FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		
		//Jab 클래스 설정
		job.setJarByClass(AvgExe.class);
		//Mapper 클래스 설정
		job.setMapperClass(AvgMapper.class);
		//Reducer 클래스 설정
		job.setReducerClass(AvgReducer.class);
		
		//입출력 데이터 포맷 설정
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		//출력키 및 출력값 유형 설정
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		if(conf.get("workType").equals("both")){
			//다수의 파일을 출력하기 위해 Setting (여기서 설정한 output이름은 reducer의 이름과 같아야 한다)
			MultipleOutputs.addNamedOutput(job, "Average", TextOutputFormat.class, Text.class, IntWritable.class);
			MultipleOutputs.addNamedOutput(job, "Sum", TextOutputFormat.class, Text.class, IntWritable.class);
		}
		
		job.waitForCompletion(true);
	
	
		return 0;
	}

}
